package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo20CheckPoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo20CheckPoint")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 使用CheckPoint之前需要先设置一个目录用于保存checkpoint
    sc.setCheckpointDir("Spark/data/checkpoint")

    val stuRDD: RDD[String] = sc.textFile("Spark/data/students.txt")

    val stuMapRDD: RDD[String] = stuRDD.map(line => {
      println("执行了依次map")
      line
    })

    /**
     * checkpoint并不能用于缓存数据
     * 一般用于SparkStreaming中保存实时任务运行的某个时刻的状态 适用于容错
     * 还可以切断RDD之间的依赖关系
     * checkPoint是可靠的，需要借助HDFS来进行存储，数据不会丢失
     * 缓存的数据是保存在Executor中，Executor很容易出现故障，不能保证数据的可靠性
     * 在进行checkPoint操作的时候会从头开始所有流程全部执行一次 所以可以在checkPoint之前做一次缓存提高效率
     */
    stuMapRDD.cache()
    stuMapRDD.checkpoint()

    // 统计班级人数
    stuMapRDD.map(line => (line.split(",")(4), 1))
      .reduceByKey(_ + _)
      .foreach(println)


    //    // 统计性别人数
    //    stuMapRDD.map(line => (line.split(",")(3), 1))
    //      .reduceByKey(_ + _)
    //      .foreach(println)

    while (true) {

    }
  }

}
